#include "GroupByOperationNode.h"
#include "HashTable.h"
#include "types.h"
#include "config.h"

GroupByOperationNode::GroupByOperationNode(OperationNode* source,
		SharedArray<int> groupedColumns, SharedArray<AggregationData> aggregations):
		_source(source),
		_groupedColumns(groupedColumns),
		_aggregations(aggregations) {}

GroupByOperationNode::~GroupByOperationNode() {
	delete _source;
}

SharedArray<Column> GroupByOperationNode::pull() {
	SharedArray<Column> in = _source->pull();
	assert(in.size() != 0);

	if (!_aggregator.isInitialized())
		_aggregator.init(in, _groupedColumns, _aggregations);

	if (!_aggregator.hasAggregatedData()) {
		HashTable htable(_aggregator.keySize(), _aggregator.valueSize(), _aggregator.defaultValue());

		_aggregator.loadInputData(in);

		while (in[0].size() != 0) {
			SharedArray<char*> valueRows = htable.get(_aggregator.concaternatedKeys(), _aggregator.hashes(), _aggregator.dataSize());

			_aggregator.aggregate(valueRows);

			in = _source->pull();
			_aggregator.loadInputData(in);
		}

		_aggregator.loadAggregatedData(htable.getEverything(), PULL_AMOUNT);
	}

	return _aggregator.takeAggregatedDataChunk();
}
